package cn.dmrliu.edu.realtime.app.dwd.log;

import cn.dmrliu.edu.realtime.util.KafkaUtil;
import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.connector.kafka.source.KafkaSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SideOutputDataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.util.Collector;
import org.apache.flink.util.OutputTag;

/**
 * 流量域的处理
 */
public class DwdTrafficBaseLogSplit {
    public static void main(String[] args) throws Exception {
        // TODO 1. 基本环境的准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);

        // TODO 2. 检查点的设置

        // TODO 3.从kafka的主题中读取数据
        String topic = "edu_log";
        String group = "edu_dwd_traffic_base_log_split_group";
        KafkaSource<String> kafkaSource = KafkaUtil.getKafkaSource(topic, group);
        DataStreamSource<String> kafkaDS = env.fromSource(kafkaSource, WatermarkStrategy.noWatermarks(), "kafka source");
//        kafkaDS.print(">>>");

        // TODO 4.数据清洗并转换格式,脏数据输出到侧输出流
        OutputTag<String> dirtyTag = new OutputTag<String>("dirty") {
        };
        SingleOutputStreamOperator<JSONObject> etlJsonObjectDS = kafkaDS.process(new ProcessFunction<String, JSONObject>() {
            @Override
            public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                try {
                    JSONObject jsonObject = JSONObject.parseObject(jsonStr);
                    out.collect(jsonObject);
                } catch (Exception e) {
                    ctx.output(dirtyTag, jsonStr);
                }
            }
        });
        SideOutputDataStream<String> dirtyDS = etlJsonObjectDS.getSideOutput(dirtyTag);
        dirtyDS.sinkTo(KafkaUtil.getKafkaSink("edu_dirty"));

        // TODO 5.分流
        OutputTag<String> errTag = new OutputTag<String>("errTag") {
        };
        OutputTag<String> actionTag = new OutputTag<String>("actionTag") {
        };
        OutputTag<String> displayTag = new OutputTag<String>("displayTag") {
        };
        OutputTag<String> startTag = new OutputTag<String>("startTag") {
        };
        OutputTag<String> videoTag = new OutputTag<String>("videoTag") {
        };

        SingleOutputStreamOperator<String> pageDS = etlJsonObjectDS.process(new ProcessFunction<JSONObject, String>() {
            @Override
            public void processElement(JSONObject jsonObject, Context ctx, Collector<String> out) throws Exception {
                // 错误信息
                JSONObject err = jsonObject.getJSONObject("err");
                if (err != null) {
                    ctx.output(errTag, jsonObject.toJSONString());//包含了错误，就要把所有信息都发送
                    jsonObject.remove("err");
                }

                if (jsonObject.getJSONObject("appVideo") != null) {
                    // 视频日志
                    ctx.output(videoTag, jsonObject.toJSONString());
                } else if (jsonObject.getJSONObject("start") != null) {
                    // 启动日志
                    ctx.output(startTag, jsonObject.toJSONString());
                } else {
                    JSONObject page = jsonObject.getJSONObject("page");
                    JSONObject common = jsonObject.getJSONObject("common");
                    Long ts = jsonObject.getLong("ts");
                    // 曝光日志
                    JSONArray displayArr = jsonObject.getJSONArray("displays");
                    if (displayArr != null && displayArr.size() > 0) {
                        for (int i = 0; i < displayArr.size(); i++) {
                            JSONObject displaysJSONObject = displayArr.getJSONObject(i);
                            JSONObject display = new JSONObject();
                            display.put("common", common);
                            display.put("ts", ts);
                            display.put("page", page);
                            display.put("display", displaysJSONObject);
                            ctx.output(displayTag, display.toJSONString());
                        }
                        jsonObject.remove("displays");
                    }

                    // 动作日志
                    JSONArray actionArr = jsonObject.getJSONArray("actions");
                    if (actionArr != null && actionArr.size() > 0) {
                        for (int i = 0; i < actionArr.size(); i++) {
                            JSONObject actionJsonObject = actionArr.getJSONObject(i);
                            JSONObject action = new JSONObject();
                            action.put("common", common);
                            action.put("ts", ts);
                            action.put("page", page);
                            action.put("action", actionJsonObject);
                            ctx.output(actionTag, action.toJSONString());
                        }
                        jsonObject.remove("actions");
                    }

                    // 剩下的就是页面日志
                    out.collect(jsonObject.toJSONString());
                }
            }
        });

        SideOutputDataStream<String> errTagDS = pageDS.getSideOutput(errTag);
        SideOutputDataStream<String> actionTagDS = pageDS.getSideOutput(actionTag);
        SideOutputDataStream<String> displayTagDS = pageDS.getSideOutput(displayTag);
        SideOutputDataStream<String> startTagDS = pageDS.getSideOutput(startTag);
        SideOutputDataStream<String> videoTagDS = pageDS.getSideOutput(videoTag);

        errTagDS.print(">>>");
        actionTagDS.print("###");
        displayTagDS.print("$$$");
        startTagDS.print("%%%");
        videoTagDS.print("+++");
        pageDS.print("@@@");

        // TODO 6.写入kafka中
        errTagDS.sinkTo(KafkaUtil.getKafkaSink("edu_dwd_traffic_err"));
        actionTagDS.sinkTo(KafkaUtil.getKafkaSink("edu_dwd_traffic_action"));
        displayTagDS.sinkTo(KafkaUtil.getKafkaSink("edu_dwd_traffic_display"));
        startTagDS.sinkTo(KafkaUtil.getKafkaSink("edu_dwd_traffic_start"));
        videoTagDS.sinkTo(KafkaUtil.getKafkaSink("edu_dwd_tool_video"));
        pageDS.sinkTo(KafkaUtil.getKafkaSink("edu_dwd_traffic_page"));

        env.execute();

    }
}
